//! SQL处理模块
//! 
//! 包含SQL解析、查询规划、优化和执行功能

pub mod parser;
pub mod planner;
pub mod optimizer;
pub mod optimizer_v2;
pub mod executor;
pub mod executor_v2;
pub mod vectorized_engine;
pub mod parallel_engine;
pub mod query_engine_v2;
pub mod execution;

#[cfg(test)]
mod optimizer_tests;

#[cfg(test)]
mod optimizer_simple_test;

use crate::{Error, Result};
use crate::sql::execution::FilterOperator;
use crate::sql::planner::{LogicalPlan, Assignment, Expression, BinaryOperator, LiteralValue};
use crate::storage::tuple::RowId;
use std::sync::Arc;

/// 查询引擎 - 负责SQL查询的完整处理流程
#[derive(Debug)]
pub struct QueryEngine {
    parser: parser::SqlParser,
    planner: planner::QueryPlanner,
    optimizer: optimizer::QueryOptimizer,
    storage_engine: Arc<parking_lot::RwLock<crate::storage::StorageEngine>>,
}

impl QueryEngine {
    /// 创建新的查询引擎
    pub fn new(storage_engine: Arc<parking_lot::RwLock<crate::storage::StorageEngine>>) -> Self {
        Self {
            parser: parser::SqlParser::new(),
            planner: planner::QueryPlanner::new(),
            optimizer: optimizer::QueryOptimizer::new(),
            storage_engine,
        }
    }
    
    /// 执行SQL查询
    pub fn execute(&mut self, sql: &str) -> Result<QueryResult> {
        println!("🔍 开始执行SQL: {}", sql);
        tracing::info!("执行SQL查询: {}", sql);

        // 1. 解析SQL
        println!("📝 开始解析SQL...");
        let parse_result = self.parser.parse(sql)?;
        println!("✅ SQL解析成功: {} 条语句", parse_result.statements.len());
        tracing::debug!("SQL解析完成: {} 条语句", parse_result.statements.len());

        // 2. 生成逻辑计划
        println!("🎯 开始生成逻辑计划...");
        let logical_plan = self.planner.create_logical_plan(parse_result.statements)?;
        println!("✅ 逻辑计划生成成功: {:?}", logical_plan);
        tracing::debug!("逻辑计划生成完成");

        // 3. 优化查询计划
        println!("⚡ 开始优化逻辑计划...");
        let optimized_plan = self.optimizer.optimize(logical_plan)?;
        println!("✅ 逻辑计划优化完成: {:?}", optimized_plan);
        tracing::debug!("查询优化完成");

        // 4. 使用火山模型执行器执行计划
        let result = match &optimized_plan {
            planner::LogicalPlan::CreateTable { table_name, columns, constraints } => {
                self.execute_create_table(table_name, columns, constraints)?
            }
            planner::LogicalPlan::Insert { table_name, columns, values } => {
                self.execute_insert(table_name, columns, values)?
            }
            planner::LogicalPlan::Update { table_name, assignments, condition } => {
                self.execute_update(table_name, assignments, condition)?
            }
            planner::LogicalPlan::Delete { table_name, condition } => {
                self.execute_delete(table_name, condition)?
            }
            _ => {
                self.execute_volcano_model(&optimized_plan)?
            }
        };
        tracing::info!("查询执行完成，返回 {} 行", result.rows.len());

        Ok(result)
    }

    /// 使用火山模型执行查询计划
    fn execute_volcano_model(&self, plan: &planner::LogicalPlan) -> Result<QueryResult> {
        use execution::*;

        // 创建执行上下文
        let page_manager = self.storage_engine.read().get_page_manager();
        let mut context = ExecutionContext::new(page_manager, 1);

        // 从StorageEngine加载表元数据到ExecutionContext
        self.load_table_metadata_to_context(&mut context)?;

        // 根据逻辑计划创建执行算子
        let mut operator = self.create_operator(plan, &mut context)?;

        // 打开算子
        operator.open(&mut context)?;

        // 收集所有结果
        let mut rows = Vec::new();
        while let Some(tuple) = operator.next(&mut context)? {
            rows.push(tuple.values);
        }

        // 关闭算子
        operator.close(&mut context)?;

        // 获取列名
        let columns = operator.schema().columns.iter()
            .map(|col| col.name.clone())
            .collect();

        Ok(QueryResult {
            rows,
            column_names: columns,
            affected_rows: 0,
            execution_time_ms: 0,
        })
    }

    /// 根据逻辑计划创建执行算子
    fn create_operator(&self, plan: &planner::LogicalPlan, context: &mut execution::ExecutionContext) -> Result<Box<dyn execution::Operator>> {
        use execution::*;
        use planner::LogicalPlan;

        match plan {
            LogicalPlan::TableScan { table_name, alias: _, projection } => {
                // 创建表扫描算子
                // 将字符串投影转换为索引投影
                let projection_indices = projection.as_ref().map(|proj| (0..proj.len()).collect::<Vec<usize>>());

                let mut table_scan = TableScanOperator::new(
                    table_name.clone(),
                    projection_indices,
                    context,
                )?;

                // 设置存储引擎引用
                table_scan.set_storage_engine(self.storage_engine.clone());

                Ok(Box::new(table_scan))
            }
            LogicalPlan::Projection { input, expressions } => {
                // 创建投影算子
                let input_operator = self.create_operator(input, context)?;
                // 将表达式转换为索引
                let projection_indices = (0..expressions.len()).collect::<Vec<usize>>();

                let projection = ProjectionOperator::new(
                    input_operator,
                    projection_indices,
                )?;
                Ok(Box::new(projection))
            }
            LogicalPlan::Filter { input, predicate } => {
                // 创建过滤算子
                let input_operator = self.create_operator(input, context)?;
                // 将表达式转换为过滤谓词
                let filter_predicate = self.convert_expression_to_filter_predicate(predicate, input)?;

                let filter = FilterOperator::new(
                    input_operator,
                    filter_predicate,
                );
                Ok(Box::new(filter))
            }
            LogicalPlan::Join { left, right, join_type, condition } => {
                // 创建连接算子
                let left_operator = self.create_operator(left, context)?;
                let right_operator = self.create_operator(right, context)?;
                // 创建连接条件
                let join_condition = execution::JoinCondition {
                    left_column: 0,
                    right_column: 0,
                    operator: execution::ComparisonOperator::Equal,
                };

                // 转换JoinType
                let execution_join_type = match join_type {
                    planner::JoinType::Inner => execution::JoinType::Inner,
                    planner::JoinType::Left => execution::JoinType::LeftOuter,
                    planner::JoinType::Right => execution::JoinType::RightOuter,
                    planner::JoinType::Full => execution::JoinType::FullOuter,
                };

                let join = HashJoinOperator::new(
                    left_operator,
                    right_operator,
                    join_condition,
                    execution_join_type,
                );
                Ok(Box::new(join))
            }
            LogicalPlan::Aggregate { input, group_expressions, aggregate_expressions } => {
                // 创建聚合算子
                let input_operator = self.create_operator(input, context)?;
                // 转换表达式为索引
                let group_by_columns = (0..group_expressions.len()).collect::<Vec<usize>>();
                let agg_expressions = aggregate_expressions.iter().enumerate().map(|(i, _)| {
                    execution::AggregateExpression {
                        function: execution::AggregateFunction::Count,
                        column_index: Some(i),
                        output_name: format!("agg_{}", i),
                    }
                }).collect::<Vec<_>>();

                let aggregation = AggregationOperator::new(
                    input_operator,
                    group_by_columns,
                    agg_expressions,
                    execution::ExecutionSchema { columns: vec![] }, // 临时schema
                );
                Ok(Box::new(aggregation))
            }
            LogicalPlan::Sort { input, expressions } => {
                // 创建排序算子
                let input_operator = self.create_operator(input, context)?;
                // 转换排序表达式为排序键
                let sort_keys = expressions.iter().enumerate().map(|(i, _)| {
                    execution::SortKey {
                        column_index: i,
                        direction: execution::SortDirection::Ascending,
                    }
                }).collect::<Vec<_>>();

                let sort = SortOperator::new(
                    input_operator,
                    sort_keys,
                    execution::ExecutionSchema { columns: vec![] }, // 临时schema
                );
                Ok(Box::new(sort))
            }
            LogicalPlan::Limit { input, count } => {
                // 创建限制算子
                let input_operator = self.create_operator(input, context)?;
                let limit_op = LimitOperator::new(
                    input_operator,
                    *count,
                    0, // offset默认为0
                    execution::ExecutionSchema { columns: vec![] }, // 临时schema
                );
                Ok(Box::new(limit_op))
            }
            _ => {
                Err(Error::query_execution("不支持的逻辑计划类型".to_string()))
            }
        }
    }
    
    /// 解释查询执行计划
    pub fn explain(&mut self, sql: &str) -> Result<String> {
        let parse_result = self.parser.parse(sql)?;
        let logical_plan = self.planner.create_logical_plan(parse_result.statements)?;
        let optimized_plan = self.optimizer.optimize(logical_plan)?;

        Ok(format!("查询执行计划:\n{:#?}", optimized_plan))
    }

    /// 从StorageEngine加载表元数据到ExecutionContext
    fn load_table_metadata_to_context(&self, context: &mut execution::ExecutionContext) -> Result<()> {
        let storage_engine = self.storage_engine.read();

        // 获取所有表名
        let table_names = storage_engine.list_tables();

        for table_name in table_names {
            if let Some(table) = storage_engine.get_table(&table_name) {
                // 从表模式创建ExecutionContext的表元数据
                let columns: Vec<execution::ColumnInfo> = table.schema().columns.iter().map(|col| {
                    execution::ColumnInfo {
                        name: col.name.clone(),
                        data_type: col.data_type.clone(),
                        nullable: col.nullable,
                        table_name: Some(table_name.clone()),
                    }
                }).collect();

                let table_metadata = execution::TableMetadata::new(
                    table_name.clone(),
                    execution::ExecutionSchema { columns }
                );

                context.add_table_metadata(table_metadata);
                tracing::debug!("已加载表元数据到ExecutionContext: {}", table_name);
            }
        }

        Ok(())
    }

    /// 检查表是否存在（统一检查机制）
    /// 解决存储引擎和元数据管理器状态不一致的问题
    fn table_exists(&self, table_name: &str) -> Result<bool> {
        let storage_engine = self.storage_engine.read();

        // 首先检查存储引擎中是否存在
        let exists_in_storage = storage_engine.get_table(table_name).is_some();

        // 然后检查元数据管理器中是否存在
        let exists_in_metadata = if let Some(metadata_manager) = storage_engine.get_metadata_manager() {
            metadata_manager.get_table_metadata(table_name).is_some()
        } else {
            false
        };

        // 如果两者状态不一致，记录警告并以存储引擎为准
        if exists_in_storage != exists_in_metadata {
            tracing::warn!(
                "表 '{}' 状态不一致: 存储引擎={}, 元数据管理器={}",
                table_name, exists_in_storage, exists_in_metadata
            );

            // 尝试同步状态
            if exists_in_storage && !exists_in_metadata {
                tracing::info!("表 '{}' 存在于存储引擎但不在元数据管理器中，可能需要重建元数据", table_name);
            } else if !exists_in_storage && exists_in_metadata {
                tracing::info!("表 '{}' 存在于元数据管理器但不在存储引擎中，可能需要清理元数据", table_name);
            }
        }

        // 以存储引擎的状态为准，因为它是数据的实际存储位置
        Ok(exists_in_storage)
    }

    /// 执行CREATE TABLE语句
    fn execute_create_table(
        &self,
        table_name: &str,
        columns: &[planner::ColumnDefinition],
        constraints: &[planner::TableConstraintDefinition],
    ) -> Result<QueryResult> {
        tracing::info!("执行CREATE TABLE: {}", table_name);

        // 统一检查表是否已存在
        if self.table_exists(table_name)? {
            return Err(Error::query_execution(format!("表 '{}' 已存在", table_name)));
        }

        // 转换列定义为存储引擎格式
        let mut storage_columns = Vec::new();
        let mut primary_key_columns = Vec::new();

        for col in columns {
            let storage_col = crate::storage::metadata::ColumnMetadata {
                column_name: col.name.clone(),
                data_type: self.convert_to_metadata_data_type(&col.data_type)?,
                nullable: col.nullable,
                default_value: col.default_value.as_ref().map(|v| crate::sql::Value::Text(v.clone())),
                auto_increment: col.auto_increment,
                comment: None,
                ordinal_position: storage_columns.len() as u32,
            };
            storage_columns.push(storage_col);

            // 检查是否为单列主键
            if col.primary_key {
                primary_key_columns.push(col.name.clone());
            }
        }

        // 处理表约束中的复合主键
        for constraint in constraints {
            if let planner::ConstraintType::PrimaryKey(pk_columns) = &constraint.constraint_type {
                // 如果已经有单列主键，则报错
                if !primary_key_columns.is_empty() {
                    return Err(Error::query_execution(
                        "表不能同时有单列主键和复合主键".to_string()
                    ));
                }
                primary_key_columns = pk_columns.clone();
                break;
            }
        }

        // 验证主键列是否存在
        for pk_col in &primary_key_columns {
            if !columns.iter().any(|col| &col.name == pk_col) {
                return Err(Error::query_execution(format!(
                    "主键列 '{}' 不存在", pk_col
                )));
            }
        }

        // 获取存储引擎的写锁
        let mut storage_engine = self.storage_engine.write();

        // 实现事务化DDL操作
        let result = self.execute_create_table_transactional(
            &mut storage_engine,
            table_name,
            &storage_columns,
            &primary_key_columns
        );

        // 如果操作失败，确保清理任何部分状态
        if result.is_err() {
            self.cleanup_failed_table_creation(&mut storage_engine, table_name);
        }

        result
    }

    /// 事务化的CREATE TABLE执行
    /// 参考PostgreSQL的原子DDL实现
    fn execute_create_table_transactional(
        &self,
        storage_engine: &mut crate::storage::StorageEngine,
        table_name: &str,
        storage_columns: &[crate::storage::metadata::ColumnMetadata],
        primary_key_columns: &[String],
    ) -> Result<QueryResult> {
        // 为了兼容现有的元数据管理器，暂时只支持单列主键
        let single_primary_key = if primary_key_columns.len() == 1 {
            Some(primary_key_columns[0].clone())
        } else if primary_key_columns.is_empty() {
            None
        } else {
            // 复合主键暂时使用第一列作为元数据主键，实际约束检查在存储层处理
            Some(primary_key_columns[0].clone())
        };

        // 第一阶段：准备存储引擎的表结构
        let mut storage_table_columns = Vec::new();
        for col in storage_columns {
            let storage_col = crate::storage::ColumnDefinition {
                name: col.column_name.clone(),
                data_type: self.convert_metadata_to_storage_data_type(&col.data_type)?,
                nullable: col.nullable,
                // 检查是否为主键列（支持复合主键）
                primary_key: primary_key_columns.contains(&col.column_name),
            };
            storage_table_columns.push(storage_col);
        }

        let table_schema = crate::storage::TableSchema::new(storage_table_columns);
        let storage_columns_clone = storage_columns.to_vec();

        // 第二阶段：原子性创建（先存储引擎，后元数据）
        // 这样如果元数据创建失败，存储引擎的表可以被清理
        storage_engine.create_table(table_name.to_string(), table_schema)
            .map_err(|e| {
                tracing::error!("存储引擎创建表 '{}' 失败: {}", table_name, e);
                e
            })?;

        // 第三阶段：创建元数据（如果失败，需要清理存储引擎中的表）
        let table_id = if let Some(metadata_manager) = storage_engine.get_metadata_manager() {
            metadata_manager.create_table(
                table_name.to_string(),
                storage_columns_clone,
                single_primary_key.clone(),
            ).map_err(|e| {
                tracing::error!("元数据管理器创建表 '{}' 失败: {}，正在清理存储引擎", table_name, e);
                // 清理存储引擎中已创建的表
                if let Err(cleanup_err) = storage_engine.drop_table(table_name) {
                    tracing::error!("清理存储引擎表 '{}' 失败: {}", table_name, cleanup_err);
                }
                e
            })?
        } else {
            return Err(Error::storage("元数据管理器未初始化".to_string()));
        };

        tracing::info!("表 '{}' 创建成功，表ID: {}，已在存储引擎中创建", table_name, table_id);

        Ok(QueryResult {
            rows: vec![],
            column_names: vec![],
            affected_rows: 1,
            execution_time_ms: 0,
        })
    }

    /// 清理失败的表创建操作
    /// 确保在CREATE TABLE失败时不留下部分状态
    fn cleanup_failed_table_creation(
        &self,
        storage_engine: &mut crate::storage::StorageEngine,
        table_name: &str,
    ) {
        tracing::warn!("清理失败的表创建操作: {}", table_name);

        // 清理存储引擎中的表
        if let Err(e) = storage_engine.drop_table(table_name) {
            tracing::debug!("清理存储引擎表时出错（可能表不存在）: {}", e);
        }

        // 清理元数据管理器中的表
        if let Some(metadata_manager) = storage_engine.get_metadata_manager() {
            if let Err(e) = metadata_manager.drop_table(table_name) {
                tracing::debug!("清理元数据表时出错（可能表不存在）: {}", e);
            }
        }

        tracing::info!("表 '{}' 清理完成", table_name);
    }

    /// 执行INSERT语句
    fn execute_insert(
        &self,
        table_name: &str,
        columns: &Option<Vec<String>>,
        values: &[Vec<planner::Expression>],
    ) -> Result<QueryResult> {
        tracing::info!("执行INSERT INTO {}", table_name);

        // 先检查表是否存在（使用读锁）
        if !self.table_exists(table_name)? {
            return Err(Error::query_execution(format!("表 '{}' 不存在", table_name)));
        }

        // 获取存储引擎的写锁
        let mut storage_engine = self.storage_engine.write();

        // 获取元数据管理器
        if let Some(metadata_manager) = storage_engine.get_metadata_manager() {
            // 获取表元数据
            let table_metadata = metadata_manager.get_table_metadata(table_name)
                .ok_or_else(|| Error::query_execution(format!("表 '{}' 元数据不存在", table_name)))?;

            // 确定要插入的列
            let target_columns = if let Some(cols) = columns {
                cols.clone()
            } else {
                // 如果没有指定列，使用表的所有列
                table_metadata.columns.iter().map(|col| col.column_name.clone()).collect()
            };

            let mut affected_rows = 0;

            // 处理每一行数据
            for value_row in values {
                if value_row.len() != target_columns.len() {
                    return Err(Error::query_execution(
                        format!("值的数量({})与列的数量({})不匹配", value_row.len(), target_columns.len())
                    ));
                }

                // 创建完整的行数据，包含所有列
                let mut full_row_values = Vec::new();

                // 为表的每一列准备值
                for table_column in &table_metadata.columns {
                    if let Some(target_index) = target_columns.iter().position(|col| col == &table_column.column_name) {
                        // 这一列在INSERT语句中指定了值
                        let expr = &value_row[target_index];
                        let value = self.evaluate_expression(expr, table_column)?;
                        full_row_values.push(value);
                    } else {
                        // 这一列没有在INSERT语句中指定，使用默认值或NULL
                        if let Some(default_value) = &table_column.default_value {
                            full_row_values.push(default_value.clone());
                        } else if table_column.nullable {
                            full_row_values.push(Value::Null);
                        } else {
                            return Err(Error::query_execution(
                                format!("列 '{}' 不能为NULL且没有默认值", table_column.column_name)
                            ));
                        }
                    }
                }

                // 实际插入数据到存储引擎
                tracing::debug!("尝试获取表 '{}' 的可变引用...", table_name);
                if let Some(table) = storage_engine.get_table_mut(table_name) {
                    tracing::debug!("成功获取表 '{}' 的可变引用，准备插入数据: {:?}", table_name, full_row_values);
                    let row_id = table.insert_row(full_row_values.clone())?;
                    tracing::debug!("成功插入行数据: {:?}, 行ID: {:?}", full_row_values, row_id);
                    affected_rows += 1;
                } else {
                    tracing::error!("错误：表 '{}' 不存在于存储引擎中", table_name);
                    return Err(Error::query_execution(format!("表 '{}' 不存在于存储引擎中", table_name)));
                }
            }

            Ok(QueryResult {
                rows: vec![],
                column_names: vec![],
                affected_rows,
                execution_time_ms: 0,
            })
        } else {
            Err(Error::storage("元数据管理器未初始化".to_string()))
        }
    }

    /// 执行UPDATE语句
    /// 支持带WHERE条件的更新操作，实现完整的UPDATE功能
    fn execute_update(
        &self,
        table_name: &str,
        assignments: &[planner::Assignment],
        condition: &Option<planner::Expression>,
    ) -> Result<QueryResult> {
        tracing::info!("执行UPDATE语句: 表={}, 赋值数={}, 有条件={}",
                      table_name, assignments.len(), condition.is_some());

        let start_time = std::time::Instant::now();

        // 先检查表是否存在（使用读锁）
        if !self.table_exists(table_name)? {
            return Err(Error::query_execution(format!("表 '{}' 不存在", table_name)));
        }

        // 获取存储引擎的写锁
        let mut storage_engine = self.storage_engine.write();

        // 获取表的元数据
        let table_metadata = if let Some(metadata_manager) = storage_engine.get_metadata_manager() {
            metadata_manager.get_table_metadata(table_name)
                .ok_or_else(|| Error::query_execution(format!("无法获取表 '{}' 的元数据", table_name)))?
        } else {
            return Err(Error::storage("元数据管理器未初始化".to_string()));
        };

        // 验证赋值操作中的列是否存在
        for assignment in assignments {
            if !table_metadata.columns.iter().any(|col| col.column_name == assignment.column) {
                return Err(Error::query_execution(format!("列 '{}' 在表 '{}' 中不存在", assignment.column, table_name)));
            }
        }

        // 获取表数据（简化实现：从内存表中获取）
        let table = storage_engine.get_table(table_name)
            .ok_or_else(|| Error::query_execution(format!("表 '{}' 不存在", table_name)))?;

        let mut affected_rows = 0;
        let column_names: Vec<String> = table_metadata.columns.iter().map(|col| col.column_name.clone()).collect();

        // 获取当前事务ID（简化实现）
        let transaction_id = u32::MAX; // 使用最大事务ID来看到所有数据

        // 实现真实的UPDATE操作
        // 1. 获取表的可变引用
        if let Some(table) = storage_engine.get_table_mut(table_name) {
            // 2. 扫描表中的所有行，获取行ID和数据
            let transaction_id = u32::MAX; // 使用最大事务ID来看到所有数据
            let rows_with_id = table.scan_rows_with_row_id(transaction_id)?;

            tracing::info!("UPDATE操作：扫描到 {} 行数据，表名: {}", rows_with_id.len(), table_name);
            println!("UPDATE操作：扫描到 {} 行数据，表名: {}", rows_with_id.len(), table_name);

            // 3. 遍历每一行，应用WHERE条件和UPDATE操作
            for (row_id, row_data) in rows_with_id {
                // 检查WHERE条件（如果存在）
                let should_update = if let Some(condition_expr) = condition {
                    self.evaluate_condition(condition_expr, &row_data, &column_names)?
                } else {
                    true // 没有WHERE条件，更新所有行
                };

                if should_update {
                    // 4. 应用赋值操作 - 计算新值
                    let mut updated_row = row_data.clone();

                    for assignment in assignments {
                        let column_index = column_names.iter()
                            .position(|col| col == &assignment.column)
                            .ok_or_else(|| Error::query_execution(format!("列 '{}' 不存在", assignment.column)))?;

                        // 计算新值
                        let new_value = self.evaluate_assignment_expression(&assignment.value, &row_data, &column_names)?;

                        // 更新行数据
                        if column_index < updated_row.len() {
                            updated_row[column_index] = new_value;
                            tracing::debug!("更新行 {:?} 列 '{}' 的值为: {:?}", row_id, assignment.column, updated_row[column_index]);
                        }
                    }

                    // 5. 执行真实的UPDATE操作
                    let update_success = table.update_row(row_id, updated_row)?;
                    if update_success {
                        affected_rows += 1;
                        tracing::debug!("成功更新行 {:?}", row_id);
                    }
                }
            }
        } else {
            return Err(Error::query_execution(format!("表 '{}' 不存在于存储引擎中", table_name)));
        }

        let execution_time = start_time.elapsed();
        tracing::info!("UPDATE语句执行完成，更新了 {} 行，耗时 {:?}", affected_rows, execution_time);

        Ok(QueryResult {
            rows: vec![],
            column_names: vec![],
            affected_rows,
            execution_time_ms: execution_time.as_millis() as u64,
        })
    }

    /// 执行DELETE语句
    /// 支持带WHERE条件的删除操作，实现完整的DELETE功能
    fn execute_delete(
        &self,
        table_name: &str,
        condition: &Option<planner::Expression>,
    ) -> Result<QueryResult> {
        tracing::info!("执行DELETE语句: 表={}, 有条件={}",
                      table_name, condition.is_some());

        let start_time = std::time::Instant::now();

        // 先检查表是否存在（使用读锁）
        if !self.table_exists(table_name)? {
            return Err(Error::query_execution(format!("表 '{}' 不存在", table_name)));
        }

        // 获取存储引擎的写锁
        let mut storage_engine = self.storage_engine.write();

        // 获取表的元数据
        let table_metadata = if let Some(metadata_manager) = storage_engine.get_metadata_manager() {
            metadata_manager.get_table_metadata(table_name)
                .ok_or_else(|| Error::query_execution(format!("无法获取表 '{}' 的元数据", table_name)))?
        } else {
            return Err(Error::storage("元数据管理器未初始化".to_string()));
        };

        // 获取表数据（简化实现：从内存表中获取）
        let table = storage_engine.get_table(table_name)
            .ok_or_else(|| Error::query_execution(format!("表 '{}' 不存在", table_name)))?;

        let mut affected_rows = 0;
        let column_names: Vec<String> = table_metadata.columns.iter().map(|col| col.column_name.clone()).collect();

        // 获取当前事务ID（简化实现）
        let transaction_id = u32::MAX; // 使用最大事务ID来看到所有数据

        // 实现真实的DELETE操作
        // 1. 获取表的可变引用
        if let Some(table) = storage_engine.get_table_mut(table_name) {
            // 2. 扫描表中的所有行，获取行ID和数据
            let rows_with_id = table.scan_rows_with_row_id(transaction_id)?;
            let mut rows_to_delete = Vec::new();

            tracing::debug!("扫描到 {} 行数据用于DELETE操作", rows_with_id.len());

            // 3. 遍历每一行，应用WHERE条件
            for (row_id, row_data) in rows_with_id {
                // 检查WHERE条件（如果存在）
                let should_delete = if let Some(condition_expr) = condition {
                    self.evaluate_condition(condition_expr, &row_data, &column_names)?
                } else {
                    true // 没有WHERE条件，删除所有行
                };

                if should_delete {
                    // 记录要删除的行ID
                    rows_to_delete.push(row_id);
                    tracing::debug!("标记删除行 {:?}: {:?}", row_id, row_data);
                }
            }

            // 4. 执行真实的删除操作
            for row_id in rows_to_delete {
                let delete_success = table.delete_row(row_id)?;
                if delete_success {
                    affected_rows += 1;
                    tracing::debug!("成功删除行 {:?}", row_id);
                }
            }
        } else {
            return Err(Error::query_execution(format!("表 '{}' 不存在于存储引擎中", table_name)));
        }

        let execution_time = start_time.elapsed();
        tracing::info!("DELETE语句执行完成，删除了 {} 行，耗时 {:?}", affected_rows, execution_time);

        Ok(QueryResult {
            rows: vec![],
            column_names: vec![],
            affected_rows,
            execution_time_ms: execution_time.as_millis() as u64,
        })
    }

    /// 评估WHERE条件表达式，返回布尔值
    fn evaluate_condition(
        &self,
        expr: &planner::Expression,
        row_data: &[Value],
        column_names: &[String],
    ) -> Result<bool> {
        let value = self.evaluate_expression_with_context(expr, row_data, column_names)?;
        match value {
            Value::Boolean(b) => Ok(b),
            Value::Null => Ok(false), // NULL条件视为false
            _ => Err(Error::query_execution("WHERE条件必须返回布尔值")),
        }
    }

    /// 评估赋值表达式，返回新值
    fn evaluate_assignment_expression(
        &self,
        expr: &planner::Expression,
        row_data: &[Value],
        column_names: &[String],
    ) -> Result<Value> {
        self.evaluate_expression_with_context(expr, row_data, column_names)
    }

    /// 在给定行上下文中计算表达式的值
    fn evaluate_expression_with_context(
        &self,
        expr: &planner::Expression,
        row_data: &[Value],
        column_names: &[String],
    ) -> Result<Value> {
        match expr {
            planner::Expression::Literal(literal) => {
                match literal {
                    planner::LiteralValue::Integer(i) => Ok(Value::Integer(*i)),
                    planner::LiteralValue::Float(f) => Ok(Value::Real(*f)),
                    planner::LiteralValue::String(s) => Ok(Value::Text(s.clone())),
                    planner::LiteralValue::Boolean(b) => Ok(Value::Boolean(*b)),
                    planner::LiteralValue::Null => Ok(Value::Null),
                }
            }
            planner::Expression::Column { name, table: _ } => {
                // 查找列在行数据中的位置
                let column_index = column_names.iter()
                    .position(|col| col == name)
                    .ok_or_else(|| Error::query_execution(format!("列 '{}' 不存在", name)))?;

                Ok(row_data.get(column_index).cloned().unwrap_or(Value::Null))
            }
            planner::Expression::BinaryOp { left, operator, right } => {
                let left_value = self.evaluate_expression_with_context(left, row_data, column_names)?;
                let right_value = self.evaluate_expression_with_context(right, row_data, column_names)?;
                self.evaluate_binary_operation(&left_value, operator, &right_value)
            }
            planner::Expression::Wildcard => {
                Err(Error::query_execution("通配符不能在此上下文中使用"))
            }
        }
    }

    /// 计算二元操作的结果
    fn evaluate_binary_operation(
        &self,
        left: &Value,
        operator: &planner::BinaryOperator,
        right: &Value,
    ) -> Result<Value> {
        use planner::BinaryOperator;

        match (left, right) {
            // 处理NULL值
            (Value::Null, _) | (_, Value::Null) => {
                match operator {
                    BinaryOperator::Equal => Ok(Value::Boolean(false)),
                    BinaryOperator::NotEqual => Ok(Value::Boolean(true)),
                    _ => Ok(Value::Null),
                }
            }

            // 整数运算
            (Value::Integer(l), Value::Integer(r)) => {
                match operator {
                    BinaryOperator::Plus => Ok(Value::Integer(l + r)),
                    BinaryOperator::Minus => Ok(Value::Integer(l - r)),
                    BinaryOperator::Multiply => Ok(Value::Integer(l * r)),
                    BinaryOperator::Divide => {
                        if *r == 0 {
                            Err(Error::query_execution("除零错误"))
                        } else {
                            Ok(Value::Integer(l / r))
                        }
                    }
                    BinaryOperator::Equal => Ok(Value::Boolean(l == r)),
                    BinaryOperator::NotEqual => Ok(Value::Boolean(l != r)),
                    BinaryOperator::LessThan => Ok(Value::Boolean(l < r)),
                    BinaryOperator::LessThanOrEqual => Ok(Value::Boolean(l <= r)),
                    BinaryOperator::GreaterThan => Ok(Value::Boolean(l > r)),
                    BinaryOperator::GreaterThanOrEqual => Ok(Value::Boolean(l >= r)),
                    _ => Err(Error::query_execution(format!("不支持的整数操作: {:?}", operator))),
                }
            }

            // 浮点数运算
            (Value::Real(l), Value::Real(r)) => {
                match operator {
                    BinaryOperator::Plus => Ok(Value::Real(l + r)),
                    BinaryOperator::Minus => Ok(Value::Real(l - r)),
                    BinaryOperator::Multiply => Ok(Value::Real(l * r)),
                    BinaryOperator::Divide => {
                        if *r == 0.0 {
                            Err(Error::query_execution("除零错误"))
                        } else {
                            Ok(Value::Real(l / r))
                        }
                    }
                    BinaryOperator::Equal => Ok(Value::Boolean((l - r).abs() < f64::EPSILON)),
                    BinaryOperator::NotEqual => Ok(Value::Boolean((l - r).abs() >= f64::EPSILON)),
                    BinaryOperator::LessThan => Ok(Value::Boolean(l < r)),
                    BinaryOperator::LessThanOrEqual => Ok(Value::Boolean(l <= r)),
                    BinaryOperator::GreaterThan => Ok(Value::Boolean(l > r)),
                    BinaryOperator::GreaterThanOrEqual => Ok(Value::Boolean(l >= r)),
                    _ => Err(Error::query_execution(format!("不支持的浮点数操作: {:?}", operator))),
                }
            }

            // 字符串运算
            (Value::Text(l), Value::Text(r)) => {
                match operator {
                    BinaryOperator::Equal => Ok(Value::Boolean(l == r)),
                    BinaryOperator::NotEqual => Ok(Value::Boolean(l != r)),
                    BinaryOperator::LessThan => Ok(Value::Boolean(l < r)),
                    BinaryOperator::LessThanOrEqual => Ok(Value::Boolean(l <= r)),
                    BinaryOperator::GreaterThan => Ok(Value::Boolean(l > r)),
                    BinaryOperator::GreaterThanOrEqual => Ok(Value::Boolean(l >= r)),
                    _ => Err(Error::query_execution(format!("不支持的字符串操作: {:?}", operator))),
                }
            }

            // 布尔运算
            (Value::Boolean(l), Value::Boolean(r)) => {
                match operator {
                    BinaryOperator::And => Ok(Value::Boolean(*l && *r)),
                    BinaryOperator::Or => Ok(Value::Boolean(*l || *r)),
                    BinaryOperator::Equal => Ok(Value::Boolean(l == r)),
                    BinaryOperator::NotEqual => Ok(Value::Boolean(l != r)),
                    _ => Err(Error::query_execution(format!("不支持的布尔操作: {:?}", operator))),
                }
            }

            // 混合类型比较（整数和浮点数）
            (Value::Integer(l), Value::Real(r)) => {
                let l_float = *l as f64;
                match operator {
                    BinaryOperator::Equal => Ok(Value::Boolean((l_float - r).abs() < f64::EPSILON)),
                    BinaryOperator::NotEqual => Ok(Value::Boolean((l_float - r).abs() >= f64::EPSILON)),
                    BinaryOperator::LessThan => Ok(Value::Boolean(l_float < *r)),
                    BinaryOperator::LessThanOrEqual => Ok(Value::Boolean(l_float <= *r)),
                    BinaryOperator::GreaterThan => Ok(Value::Boolean(l_float > *r)),
                    BinaryOperator::GreaterThanOrEqual => Ok(Value::Boolean(l_float >= *r)),
                    _ => Err(Error::query_execution(format!("不支持的混合类型操作: {:?}", operator))),
                }
            }
            (Value::Real(l), Value::Integer(r)) => {
                let r_float = *r as f64;
                match operator {
                    BinaryOperator::Equal => Ok(Value::Boolean((l - r_float).abs() < f64::EPSILON)),
                    BinaryOperator::NotEqual => Ok(Value::Boolean((l - r_float).abs() >= f64::EPSILON)),
                    BinaryOperator::LessThan => Ok(Value::Boolean(*l < r_float)),
                    BinaryOperator::LessThanOrEqual => Ok(Value::Boolean(*l <= r_float)),
                    BinaryOperator::GreaterThan => Ok(Value::Boolean(*l > r_float)),
                    BinaryOperator::GreaterThanOrEqual => Ok(Value::Boolean(*l >= r_float)),
                    _ => Err(Error::query_execution(format!("不支持的混合类型操作: {:?}", operator))),
                }
            }

            // 其他类型组合
            _ => Err(Error::query_execution(format!("不支持的操作数类型组合: {:?} {:?} {:?}", left, operator, right))),
        }
    }



    /// 计算表达式的值（兼容旧接口）
    fn evaluate_expression(
        &self,
        expr: &planner::Expression,
        _column_metadata: &crate::storage::metadata::ColumnMetadata,
    ) -> Result<crate::sql::Value> {
        match expr {
            planner::Expression::Literal(literal) => {
                match literal {
                    planner::LiteralValue::Integer(i) => Ok(crate::sql::Value::Integer(*i)),
                    planner::LiteralValue::Float(f) => Ok(crate::sql::Value::Real(*f)),
                    planner::LiteralValue::String(s) => Ok(crate::sql::Value::Text(s.clone())),
                    planner::LiteralValue::Boolean(b) => Ok(crate::sql::Value::Boolean(*b)),
                    planner::LiteralValue::Null => Ok(crate::sql::Value::Null),
                }
            }
            _ => Err(Error::query_execution("不支持的表达式类型")),
        }
    }

    /// 转换数据类型到元数据格式
    fn convert_to_metadata_data_type(&self, data_type: &planner::DataType) -> Result<crate::storage::metadata::DataType> {
        match data_type {
            planner::DataType::Integer | planner::DataType::Serial => Ok(crate::storage::metadata::DataType::Integer),
            planner::DataType::BigInt => Ok(crate::storage::metadata::DataType::BigInt),
            planner::DataType::Real | planner::DataType::Float => Ok(crate::storage::metadata::DataType::Real),
            planner::DataType::Double => Ok(crate::storage::metadata::DataType::Double),
            planner::DataType::Varchar(len) => {
                Ok(crate::storage::metadata::DataType::Varchar(len.unwrap_or(255) as u32))
            }
            planner::DataType::Text | planner::DataType::String => {
                Ok(crate::storage::metadata::DataType::Text)
            }
            planner::DataType::Boolean => Ok(crate::storage::metadata::DataType::Boolean),
            planner::DataType::Date => Ok(crate::storage::metadata::DataType::Date),
            planner::DataType::Timestamp => Ok(crate::storage::metadata::DataType::Timestamp),
            planner::DataType::Decimal(precision) => Ok(crate::storage::metadata::DataType::Decimal(*precision)),
        }
    }

    /// 转换元数据数据类型到存储引擎格式
    fn convert_metadata_to_storage_data_type(&self, data_type: &crate::storage::metadata::DataType) -> Result<crate::storage::DataType> {
        match data_type {
            crate::storage::metadata::DataType::Integer => Ok(crate::storage::DataType::Integer),
            crate::storage::metadata::DataType::BigInt => Ok(crate::storage::DataType::BigInt),
            crate::storage::metadata::DataType::Real => Ok(crate::storage::DataType::Real),
            crate::storage::metadata::DataType::Double => Ok(crate::storage::DataType::Double),
            crate::storage::metadata::DataType::Varchar(len) => Ok(crate::storage::DataType::Text(Some(*len as usize))),
            crate::storage::metadata::DataType::Text => Ok(crate::storage::DataType::Text(None)),
            crate::storage::metadata::DataType::Boolean => Ok(crate::storage::DataType::Boolean),
            crate::storage::metadata::DataType::Date => Ok(crate::storage::DataType::Date),
            crate::storage::metadata::DataType::Timestamp => Ok(crate::storage::DataType::Timestamp),
            crate::storage::metadata::DataType::Decimal(precision) => Ok(crate::storage::DataType::Decimal(*precision)),
            _ => Err(Error::query_execution(format!("不支持的数据类型: {:?}", data_type))),
        }
    }

    /// 将表达式转换为过滤谓词
    fn convert_expression_to_filter_predicate(
        &self,
        expr: &Expression,
        input_plan: &LogicalPlan,
    ) -> Result<crate::sql::execution::FilterPredicate> {
        match expr {
            Expression::BinaryOp { left, operator, right } => {
                // 处理二元比较操作
                match (left.as_ref(), right.as_ref()) {
                    (Expression::Column { name, .. }, Expression::Literal(literal)) => {
                        // 列与字面值比较
                        let column_index = self.get_column_index(name, input_plan)?;
                        let comparison_op = self.convert_binary_op_to_comparison_op(operator)?;
                        let value = self.convert_literal_to_value(literal);

                        Ok(crate::sql::execution::FilterPredicate::Comparison {
                            left_column: column_index,
                            operator: comparison_op,
                            right_value: value,
                        })
                    }
                    (Expression::Literal(literal), Expression::Column { name, .. }) => {
                        // 字面值与列比较（需要反转操作符）
                        let column_index = self.get_column_index(name, input_plan)?;
                        let comparison_op = self.reverse_comparison_op(
                            &self.convert_binary_op_to_comparison_op(operator)?
                        );
                        let value = self.convert_literal_to_value(literal);

                        Ok(crate::sql::execution::FilterPredicate::Comparison {
                            left_column: column_index,
                            operator: comparison_op,
                            right_value: value,
                        })
                    }
                    (Expression::Column { name: left_name, .. }, Expression::Column { name: right_name, .. }) => {
                        // 列与列比较
                        let left_column = self.get_column_index(left_name, input_plan)?;
                        let right_column = self.get_column_index(right_name, input_plan)?;
                        let comparison_op = self.convert_binary_op_to_comparison_op(operator)?;

                        Ok(crate::sql::execution::FilterPredicate::ColumnComparison {
                            left_column,
                            operator: comparison_op,
                            right_column,
                        })
                    }
                    _ => Err(Error::query_execution("不支持的过滤表达式类型")),
                }
            }
            _ => Err(Error::query_execution("不支持的过滤表达式")),
        }
    }

    /// 获取列在表中的索引
    fn get_column_index(&self, column_name: &str, plan: &LogicalPlan) -> Result<usize> {
        match plan {
            LogicalPlan::TableScan { table_name, .. } => {
                // 从存储引擎获取表的列信息
                let storage_engine = self.storage_engine.read();
                let table = storage_engine.get_table(table_name)
                    .ok_or_else(|| Error::TableNotFound(table_name.clone()))?;

                let schema = table.schema();
                for (index, column) in schema.columns.iter().enumerate() {
                    if column.name == column_name {
                        return Ok(index);
                    }
                }
                Err(Error::ColumnNotFound(column_name.to_string()))
            }
            _ => Err(Error::query_execution("无法确定列索引")),
        }
    }

    /// 转换二元操作符为比较操作符
    fn convert_binary_op_to_comparison_op(&self, op: &BinaryOperator) -> Result<crate::sql::execution::ComparisonOperator> {
        match op {
            BinaryOperator::Equal => Ok(crate::sql::execution::ComparisonOperator::Equal),
            BinaryOperator::NotEqual => Ok(crate::sql::execution::ComparisonOperator::NotEqual),
            BinaryOperator::LessThan => Ok(crate::sql::execution::ComparisonOperator::LessThan),
            BinaryOperator::LessThanOrEqual => Ok(crate::sql::execution::ComparisonOperator::LessThanOrEqual),
            BinaryOperator::GreaterThan => Ok(crate::sql::execution::ComparisonOperator::GreaterThan),
            BinaryOperator::GreaterThanOrEqual => Ok(crate::sql::execution::ComparisonOperator::GreaterThanOrEqual),
            _ => Err(Error::query_execution(format!("不支持的比较操作符: {:?}", op))),
        }
    }

    /// 反转比较操作符（用于处理字面值在左侧的情况）
    fn reverse_comparison_op(&self, op: &crate::sql::execution::ComparisonOperator) -> crate::sql::execution::ComparisonOperator {
        match op {
            crate::sql::execution::ComparisonOperator::LessThan => crate::sql::execution::ComparisonOperator::GreaterThan,
            crate::sql::execution::ComparisonOperator::LessThanOrEqual => crate::sql::execution::ComparisonOperator::GreaterThanOrEqual,
            crate::sql::execution::ComparisonOperator::GreaterThan => crate::sql::execution::ComparisonOperator::LessThan,
            crate::sql::execution::ComparisonOperator::GreaterThanOrEqual => crate::sql::execution::ComparisonOperator::LessThanOrEqual,
            crate::sql::execution::ComparisonOperator::Equal => crate::sql::execution::ComparisonOperator::Equal,
            crate::sql::execution::ComparisonOperator::NotEqual => crate::sql::execution::ComparisonOperator::NotEqual,
            crate::sql::execution::ComparisonOperator::Like => crate::sql::execution::ComparisonOperator::Like,
            crate::sql::execution::ComparisonOperator::In => crate::sql::execution::ComparisonOperator::In,
        }
    }

    /// 转换字面值为值
    fn convert_literal_to_value(&self, literal: &LiteralValue) -> crate::sql::Value {
        match literal {
            LiteralValue::Integer(i) => crate::sql::Value::Integer(*i),
            LiteralValue::Float(f) => crate::sql::Value::Real(*f),
            LiteralValue::String(s) => crate::sql::Value::Text(s.clone()),
            LiteralValue::Boolean(b) => crate::sql::Value::Boolean(*b),
            LiteralValue::Null => crate::sql::Value::Null,
        }
    }
}

impl Default for QueryEngine {
    fn default() -> Self {
        // 创建默认的存储引擎配置
        let config = crate::config::DatabaseConfig::default();
        let storage_engine = Arc::new(parking_lot::RwLock::new(crate::storage::StorageEngine::new(&config).unwrap()));
        Self::new(storage_engine)
    }
}

/// 查询结果
#[derive(Debug, Clone)]
pub struct QueryResult {
    /// 结果数据行
    rows: Vec<Vec<Value>>,
    /// 列名
    column_names: Vec<String>,
    /// 受影响的行数
    affected_rows: usize,
    /// 执行时间 (毫秒)
    execution_time_ms: u64,
}

impl QueryResult {
    /// 创建新的查询结果
    pub fn new(rows: Vec<Vec<Value>>, column_names: Vec<String>, affected_rows: usize, execution_time_ms: u64) -> Self {
        Self {
            rows,
            column_names,
            affected_rows,
            execution_time_ms,
        }
    }
    
    /// 创建空结果
    pub fn empty() -> Self {
        Self {
            rows: Vec::new(),
            column_names: Vec::new(),
            affected_rows: 0,
            execution_time_ms: 0,
        }
    }

    /// 创建仅包含受影响行数的结果
    pub fn affected_rows(count: usize, execution_time_ms: u64) -> Self {
        Self {
            rows: Vec::new(),
            column_names: Vec::new(),
            affected_rows: count,
            execution_time_ms,
        }
    }

    /// 获取结果行
    pub fn rows(&self) -> &[Vec<Value>] {
        &self.rows
    }

    /// 获取列名
    pub fn column_names(&self) -> &[String] {
        &self.column_names
    }
    
    /// 获取受影响的行数
    pub fn get_affected_rows(&self) -> usize {
        self.affected_rows
    }
    
    /// 获取总行数
    pub fn row_count(&self) -> usize {
        self.rows.len()
    }

    /// 获取列数
    pub fn column_count(&self) -> usize {
        self.column_names.len()
    }
    
    /// 获取执行时间
    pub fn execution_time_ms(&self) -> u64 {
        self.execution_time_ms
    }
    
    /// 检查是否为空结果
    pub fn is_empty(&self) -> bool {
        self.rows.is_empty()
    }
    
    /// 转换为JSON格式
    pub fn to_json(&self) -> Result<String> {
        if self.rows.is_empty() {
            return Ok(format!(
                r#"{{"affected_rows": {}, "execution_time_ms": {}}}"#,
                self.affected_rows, self.execution_time_ms
            ));
        }

        // 简化的JSON转换，实际实现需要更复杂的逻辑
        Ok(format!(
            r#"{{"row_count": {}, "column_count": {}, "execution_time_ms": {}}}"#,
            self.row_count(),
            self.column_count(),
            self.execution_time_ms
        ))
    }
}

/// 数据值类型
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub enum Value {
    Null,
    Boolean(bool),
    Integer(i64),
    Real(f64),
    Text(String),
    Timestamp(chrono::DateTime<chrono::Utc>),
    Decimal(rust_decimal::Decimal),
}

// 手动实现PartialEq，处理浮点数比较
impl PartialEq for Value {
    fn eq(&self, other: &Self) -> bool {
        match (self, other) {
            (Value::Null, Value::Null) => true,
            (Value::Boolean(a), Value::Boolean(b)) => a == b,
            (Value::Integer(a), Value::Integer(b)) => a == b,
            (Value::Real(a), Value::Real(b)) => {
                // 使用位比较来处理NaN和特殊值
                a.to_bits() == b.to_bits()
            }
            (Value::Text(a), Value::Text(b)) => a == b,
            (Value::Timestamp(a), Value::Timestamp(b)) => a == b,
            (Value::Decimal(a), Value::Decimal(b)) => a == b,
            _ => false,
        }
    }
}

// 实现Eq trait
impl Eq for Value {}

// 手动实现PartialOrd，处理浮点数排序
impl PartialOrd for Value {
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
        Some(self.cmp(other))
    }
}

// 手动实现Ord，定义完整的排序规则
impl Ord for Value {
    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
        use std::cmp::Ordering;

        match (self, other) {
            // Null值最小
            (Value::Null, Value::Null) => Ordering::Equal,
            (Value::Null, _) => Ordering::Less,
            (_, Value::Null) => Ordering::Greater,

            // 同类型比较
            (Value::Boolean(a), Value::Boolean(b)) => a.cmp(b),
            (Value::Integer(a), Value::Integer(b)) => a.cmp(b),
            (Value::Real(a), Value::Real(b)) => {
                // 处理浮点数比较，包括NaN
                match (a.is_nan(), b.is_nan()) {
                    (true, true) => Ordering::Equal,
                    (true, false) => Ordering::Greater, // NaN最大
                    (false, true) => Ordering::Less,
                    (false, false) => a.partial_cmp(b).unwrap_or(Ordering::Equal),
                }
            }
            (Value::Text(a), Value::Text(b)) => a.cmp(b),
            (Value::Timestamp(a), Value::Timestamp(b)) => a.cmp(b),
            (Value::Decimal(a), Value::Decimal(b)) => a.cmp(b),

            // 不同类型按类型优先级排序
            _ => self.type_priority().cmp(&other.type_priority()),
        }
    }
}

// 实现Hash trait
impl std::hash::Hash for Value {
    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
        match self {
            Value::Null => 0u8.hash(state),
            Value::Boolean(b) => {
                1u8.hash(state);
                b.hash(state);
            }
            Value::Integer(i) => {
                2u8.hash(state);
                i.hash(state);
            }
            Value::Real(f) => {
                3u8.hash(state);
                f.to_bits().hash(state); // 使用位表示来hash浮点数
            }
            Value::Text(s) => {
                4u8.hash(state);
                s.hash(state);
            }
            Value::Timestamp(ts) => {
                5u8.hash(state);
                ts.hash(state);
            }
            Value::Decimal(d) => {
                6u8.hash(state);
                d.hash(state);
            }
        }
    }
}

impl Value {
    /// 获取类型优先级（用于不同类型间的排序）
    fn type_priority(&self) -> u8 {
        match self {
            Value::Null => 0,
            Value::Boolean(_) => 1,
            Value::Integer(_) => 2,
            Value::Real(_) => 3,
            Value::Decimal(_) => 4,
            Value::Text(_) => 5,
            Value::Timestamp(_) => 6,
        }
    }

    /// 获取值的类型名称
    pub fn type_name(&self) -> &'static str {
        match self {
            Value::Null => "null",
            Value::Boolean(_) => "boolean",
            Value::Integer(_) => "integer",
            Value::Real(_) => "real",
            Value::Decimal(_) => "decimal",
            Value::Text(_) => "text",
            Value::Timestamp(_) => "timestamp",
        }
    }

    /// 检查是否为null
    pub fn is_null(&self) -> bool {
        matches!(self, Value::Null)
    }

    /// 转换为字符串表示
    pub fn to_string_repr(&self) -> String {
        match self {
            Value::Null => "NULL".to_string(),
            Value::Boolean(b) => b.to_string(),
            Value::Integer(i) => i.to_string(),
            Value::Real(f) => f.to_string(),
            Value::Decimal(d) => d.to_string(),
            Value::Text(s) => s.clone(),
            Value::Timestamp(ts) => ts.to_rfc3339(),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    
    #[tokio::test]
    async fn test_query_engine_creation() {
        let config = crate::config::DatabaseConfig::default();
        let storage_engine = Arc::new(parking_lot::RwLock::new(crate::storage::StorageEngine::new(&config).unwrap()));
        let engine = QueryEngine::new(storage_engine);
        assert!(true); // 基础创建测试
    }
    
    #[test]
    fn test_query_result_creation() {
        let result = QueryResult::empty();
        assert!(result.is_empty());
        assert_eq!(result.row_count(), 0);
        assert_eq!(result.column_count(), 0);
        
        let result = QueryResult::affected_rows(5, 100);
        assert_eq!(result.get_affected_rows(), 5);
        assert_eq!(result.execution_time_ms(), 100);
    }
    
    #[test]
    fn test_query_result_json() {
        let result = QueryResult::affected_rows(3, 50);
        let json = result.to_json().unwrap();
        assert!(json.contains("affected_rows"));
        assert!(json.contains("3"));
        assert!(json.contains("50"));
    }
}
